package com.jml.mapreduce.计数器数据清洗;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * ● 需求
 * ：
 *   ○ 去除日志中字段个数小于等于11的日志。
 *   ○ 期望输出数据
 * ：每行字段长度都大于11。
 * ● 需求分析
 * ：
 *   ○ 需要在Map阶段对输入的数据根据规则进行过滤清洗。
 */
public class ETLDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();

        //往yarn上提交需要添加得到配置。
        /*configuration.set("fs.defaultFS","hdfs://hadoop101:8020");
        configuration.set("mapreduce.framework.name","yarn");
        configuration.set("mapreduce.app-submission.cross-platform","true");
        configuration.set("yarn.resourcemanager.hostname","hadoop102");*/

        Job job = Job.getInstance(configuration);

        job.setJarByClass(ETLDriver.class);

        job.setMapperClass(ETLMapper.class);
        job.setNumReduceTasks(0);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        FileInputFormat.setInputPaths(job, new Path("D:\\tmp\\bigDate\\input001"));
        FileOutputFormat.setOutputPath(job, new Path("D:\\tmp\\bigDate\\output003"));

        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}
